/**
 * Copyright 2023 Huawei Cloud Computing Technologies Co., Ltd.

 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import { DeepstreamPlugin, DeepstreamServices, EVENT } from '@deepstream/types';
import {
    OBSResponse, InterfaceResult, GetObjectMetadataResponseContent,
} from 'esdk-obs-nodejs';
import * as OBSClient from 'esdk-obs-nodejs';
import { existsSync, promises } from 'fs';
import {
    isAbsolute, join, resolve, sep,
} from 'path';
import { SyncDirObserver } from './directory-operate';
import AsyncQueue from './async-queue';

const { stat } = promises;

/**
 * Handle OBSResponse error.
 */
export class OBSError extends Error {
    constructor(public response: OBSResponse) {
        super();
        this.message = response.CommonMsg.Message || '';
    }
}

interface HuaweiCloudCredentials {
    accessKeyId: string,
    secretAccessKey: string,
}

export interface OBSSyncOptions {
    syncDir: string,
    bucketName: string,
    obsEndpoint: string,
    obsKeyPrefix?: string,
}

export default class OBSSync extends DeepstreamPlugin {
    public description = 'OBS Folder Sync';
    private logger = this.services.logger.getNameSpace('OBS_FOLDER_SYNC');
    private obsClient: OBSClient;
    private syncDir = '';
    private syncDirObserver: SyncDirObserver | null = null;
    private fileListToBeSync: Set<string> = new Set();
    private asyncQueue = new AsyncQueue(10);

    constructor(
        private options: OBSSyncOptions,
        private services: Readonly<DeepstreamServices>,
        huaweiCloudCredentials: HuaweiCloudCredentials,
    ) {
        super();
        this.obsClient = new OBSClient({
            access_key_id: huaweiCloudCredentials.accessKeyId, // 配置AK
            secret_access_key: huaweiCloudCredentials.secretAccessKey, // 配置SK
            server: options.obsEndpoint, // 配置服务地址
            timeout: 60,
            ssl_verify: false,
            long_conn_param: 0,
        });
    }

    public init() {
        if (typeof this.options.bucketName !== 'string') {
            this.logger.fatal(EVENT.ERROR, 'Invalid or missing "bucketName"');
        }
        if (typeof this.options.obsEndpoint !== 'string') {
            this.logger.fatal(EVENT.ERROR, 'Invalid or missing "obsEndpoint"');
        }
        if (typeof this.options.syncDir !== 'string') {
            this.logger.fatal(EVENT.ERROR, 'Invalid or missing "syncDir"');
        }
        this.syncDir = this.options.syncDir;
        if (!isAbsolute(this.syncDir)) {
            this.syncDir = resolve(__dirname, this.syncDir);
        }
        if (existsSync(this.syncDir) === false) {
            this.logger.fatal(EVENT.ERROR, 'Sync directory doesn\'t exist');
        }
        if (this.options.obsKeyPrefix?.match(/^[/\\]/)) {
            this.logger.fatal(EVENT.ERROR, 'Invalid "obsKeyPrefix"');
        }
    }

    public async whenReady(): Promise<void> {
        try {
            if (await this.isBucketExist()) {
                this.logger.info(EVENT.INFO, `Huawei Cloud Bucket '${this.options.bucketName}' exists`);
            } else {
                this.logger.fatal(
                    EVENT.ERROR,
                    `Huawei Cloud Bucket ${this.options.bucketName} does not exist`,
                );
            }
        } catch (error) {
            this.logger.fatal(
                EVENT.ERROR,
                `Huawei Cloud OBS service error caught: ${(error as Error).message || 'Unknown error'}`,
            );
        }
        this.sync();
    }

    public async close(): Promise<void> {
        await this.syncDirObserver?.close();
    }

    /**
     * Sync directory
     */
    private sync() {
        this.syncDirObserver = new SyncDirObserver(this.syncDir, this.logger);
        this.syncDirObserver.onFileAdd((fileList) => {
            for (let i = 0; i < fileList.length; i++) {
                this.fileListToBeSync.add(fileList[i]);
            }
            this.putInQueue();
        });
    }

    /**
     * Put objects in OBS service queue
     */
    private async putInQueue() {
        while (this.fileListToBeSync.size) {
            let values = this.fileListToBeSync.values();
            let next = values.next();
            while (next && !next.done) {
                const filename = next.value;
                this.fileListToBeSync.delete(filename);
                // Put the job in asynchronous concurrent queue
                // Notice: asyncDoJob does NOT await for the job, but for the asynchronous concurrent queue
                await this.asyncQueue.asyncDoJob(async () => {
                    const fullPath = resolve(this.syncDir, filename);
                    try {
                        await this.putNewerObject(fullPath, filename);
                    } catch (error) {
                        this.logger.fatal(
                            EVENT.ERROR,
                            `Huawei Cloud OBS service error caught: ${(error as Error).message || 'Unknown error'}`,
                        );
                    }
                });
                next = values.next();
            }
            values = this.fileListToBeSync.values();
        }
    }

    /**
    * Check bucket exist
    * @returns true / false
    */
    private async isBucketExist() {
        const result = await this.obsClient.headBucket({
            Bucket: this.options.bucketName,
        });
        if (result.CommonMsg.Status < 300) {
            return true;
        }
        if (result.CommonMsg.Status === 404) {
            return false;
        }
        const error = new OBSError(result);
        return Promise.reject(error);
    }

    /**
     * Upload object if it's newer
     * @param sourceFile
     * @param objectKey
     * @returns
     */
    async putNewerObject(sourceFile: string, objectKey: string) {
        let metaData: InterfaceResult<GetObjectMetadataResponseContent> | null = null;
        try {
            metaData = await this.getObjectMetadata(objectKey);
        } catch (error) {
            if (error instanceof OBSError && error.response.CommonMsg.Status === 404) {
                // The file not in OBS, nothing to do
            } else {
                // Error throws
                throw error;
            }
        }
        if (metaData && (await stat(sourceFile)).mtimeMs < new Date(metaData.LastModified).getTime()) {
            // Local file is older, nothing to do.
        } else {
            await this.putObject(sourceFile, objectKey);
        }
    }

    /**
     * Upload object
     * @param sourceFile
     * @param objectKey
     * @returns
     */
    private async putObject(sourceFile: string, objectKey: string) {
        const result = await this.obsClient.putObject({
            Bucket: this.options.bucketName,
            SourceFile: sourceFile,
            Key: this.getFullObjectKey(objectKey),
        });
        if (result.CommonMsg.Status < 300) {
            return result.InterfaceResult;
        }
        const error = new OBSError(result);
        return Promise.reject(error);
    }

    /**
     * Get object metadata
     * @param objectKey
     * @returns
     */
    private async getObjectMetadata(objectKey: string) {
        const result = await this.obsClient.getObjectMetadata({
            Bucket: this.options.bucketName,
            Key: this.getFullObjectKey(objectKey),
        });
        if (result.CommonMsg.Status < 300) {
            return result.InterfaceResult;
        }
        const error = new OBSError(result);
        return Promise.reject(error);
    }

    /**
     * Combine obsKeyPrefix with object key
     * @param objectKey
     * @returns
     */
    private getFullObjectKey(objectKey: string) {
        let fullKey = join(this.options.obsKeyPrefix || '', objectKey);
        if (sep !== '/') {
            fullKey = fullKey.replace(new RegExp(`\\${sep}`, 'g'), '/');
        }
        return fullKey;
    }
}
